/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.agent.tools;

import java.io.File;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;

/**
 *
 * @author root
 */
public class HdfsTools {

    private static Logger logger = Logger.getLogger(HdfsTools.class);

    public boolean put(String localFilePath, String hdfsPath, String subjobid) {
        String fileName = "";
        String[] tmplist = localFilePath.split("/");
        fileName = tmplist[tmplist.length - 1];
        if (fileName.equals("")) {
            logger.info(subjobid + " file name is error!");
            return false;
        }
        String dst = hdfsPath;
        //Configuration conf = new Configuration();
        try {
            Configuration conf = new Configuration();
            //conf.set("dfs.socket.timeout","10");
            //conf.set("dfs.datanode.socket.write.timeout","10");
            //conf.set("fs.s3.maxRetries","1");
            conf.set("dfs.client.failover.connection.retries", "1");
            conf.set("dfs.client.failover.connection.retries.on.timeouts", "1");
            FileSystem fs = FileSystem.get(URI.create(dst), conf);
            Path srcPath = new Path(localFilePath);
            Path dstPath = new Path(hdfsPath);
            //check hdfsPath if is a Directory
            if (fs.isDirectory(dstPath)) {
                //if (fs.exists(dstPath)) {
                ;
            } else {
                fs.mkdirs(dstPath);
            }

            fs.copyFromLocalFile(srcPath, dstPath);
            logger.info(subjobid + " " + localFilePath + "upload sucess!");
            //文件原地址 
            File oldFile = new File(localFilePath);
            File oldFileOk = new File(localFilePath + ".ok");
            //文件新（目标）地址 
            String newPath = "";
            newPath = localFilePath.substring(0, localFilePath.indexOf(fileName));
            newPath = newPath + "upload_sucess/";
            //String newPath = "c:/test/"; 
            //new一个新文件夹 
            File fnewpath = new File(newPath);
            //判断文件夹是否存在 
            if (!fnewpath.exists()) {
                fnewpath.mkdirs();
            }
            //将文件移到新文件夹里 
            File fnew = new File(newPath + oldFile.getName());
            File fnewOk = new File(newPath + oldFileOk.getName());
            oldFile.renameTo(fnew);
            oldFileOk.renameTo(fnewOk);

        } catch (Exception ex) {
            logger.error(subjobid + " " + localFilePath + " hdfs put error", ex);
            return false;
        }

        return true;
    }

    public boolean getAndUploaddwd(String name, String task_id, String operator_id, String hdfsPath, String localPath) {
        String dst = hdfsPath;
        try {
            Configuration conf = new Configuration();
            FileSystem fs = FileSystem.get(URI.create(dst), conf);
            FileStatus[] fileStatusList = fs.listStatus(new Path(hdfsPath));

            List<String> txtList = new ArrayList<String>();
            List<String> okList = new ArrayList<String>();
            List<String> dwdList = new ArrayList<String>();
            for (int i = 0; i < fileStatusList.length; i++) {
                if (fileStatusList[i].isDir()) {
                    continue;
                }
                String fileName = fileStatusList[i].getPath().getName();
                if (fileName != null) {
                    if (fileName.endsWith(".ok")) {
                        okList.add(fileName);
                    } else if (fileName.endsWith(".dwd")) {
                        dwdList.add(fileName);
                    } else {
                        txtList.add(fileName);
                    }
                }
            }

            List<String> finalList = new ArrayList<String>();
            for (int i = 0; i < txtList.size(); i++) {
                String tmpOkFileName = txtList.get(i) + ".ok";
                for (int j = 0; j < okList.size(); j++) {
                    if (tmpOkFileName.equals(okList.get(j))) {
                        String tmpdwdFileName = txtList.get(i) + ".dwd";
                        boolean dwdflag = false;
                        for (int k = 0; k < dwdList.size(); k++) {
                            if (tmpdwdFileName.equals(dwdList.get(k))) {
                                dwdflag = true;//
                                break;
                            }
                        }
                        if (dwdflag) {
                            ;//don't download
                        } else {
                            finalList.add(txtList.get(i));
                        }
                        break;
                    }
                }

            }

            for (int i = 0; i < finalList.size(); i++) {
                logger.info(task_id + " " + finalList.get(i) + " begin download!");
                boolean downloadflag = get(hdfsPath, hdfsPath + "/" + finalList.get(i), localPath + "/" + finalList.get(i));//download
                if (downloadflag) {
                    File localokfile = new File(localPath + "/" + finalList.get(i) + ".ok");
                    localokfile.createNewFile();
                    boolean createFileFlag = createdwd(hdfsPath, hdfsPath + "/" + finalList.get(i));//创建一个.dwd文件标识已经下载

                    logger.info(task_id + " " + "name=" + name + ": get success!");

                } else {

                    logger.info(task_id + " " + "name=" + name + ": get err!");
                }
            }

        } catch (Exception ex) {

            logger.error(task_id + " " + hdfsPath + "getAndUploaddwd err!" + ex.getLocalizedMessage() + "|" + ex.getMessage() + "|" + ex.getStackTrace() + ex, ex);
            return false;
        }
        return true;
    }

    public boolean get(String hdfsPath, String srcFile, String dstFile) {
        String dst = hdfsPath;
        try {
            Configuration conf = new Configuration();
            FileSystem fs = FileSystem.get(URI.create(dst), conf);
            fs.copyToLocalFile(new Path(srcFile), new Path(dstFile));
        } catch (Exception ex) {
            logger.error(srcFile + " download err!" + ex, ex);
            return false;
        }

        return true;
    }

    public boolean createdwd(String hdfsPath, String srcFile) {
        String dst = hdfsPath;
        try {
            Configuration conf = new Configuration();
            FileSystem fs = FileSystem.get(URI.create(dst), conf);
            fs.createNewFile(new Path(srcFile + ".dwd"));
        } catch (Exception ex) {
            logger.error(srcFile + " createdwd err!" + ex, ex);
            return false;
        }

        return true;
    }
}
